package com.atguigu.userprofile.app;

import com.atguigu.userprofile.common.bean.TagInfo;
import com.atguigu.userprofile.common.bean.TaskInfo;
import com.atguigu.userprofile.common.bean.TaskTagRule;
import com.atguigu.userprofile.common.dao.TagInfoDAO;
import com.atguigu.userprofile.common.dao.TaskInfoDAO;
import com.atguigu.userprofile.common.dao.TaskTagRuleDAO;
import com.atguigu.userprofile.common.util.MyPropertiesUtil;
import com.atguigu.userprofile.common.util.MySQLUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

import static com.atguigu.userprofile.common.constants.ConstCode.*;

public class TaskSQL {

    //1 、 读取对应的标签定义、规则、sql
    //                查询mysql  jdbc
    //2、  建表问题
    //          最好不要手动建表 create table
    //
    //3、  查询电商离线数仓
    //                拼接组合 sql     --》 insert overwrite select ....
    //4、  生成标签信息，写入画像库
    public static void main(String[] args) {

        //spark 环境
         //System.setProperty("hadoop.home.dir", "d:\\hadoop");//打包时注释掉
         SparkConf sparkConf = new SparkConf().setAppName("task_sql_app");//.setMaster("local[*]");
         SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();


        //1 、 读取对应的标签定义、规则、sql
        //                查询mysql  jdbc
        //约定：所有经过远程提交器提交的任务 主函数的args会放两个参数
        // a task_id (task_info的主键，可以通过tag_info中的tag_task_id外键进行查询标签)
        // b busi_date (计算的业务日期，同上游的数仓一致)
        String taskId=args[0];
        String busiDate=args[1];
        //  要查哪些表？ tag_info , task_info,task_tag_rule

        TagInfo tagInfo = TagInfoDAO.getTagInfoByTaskId(taskId);

        TaskInfo taskInfo = TaskInfoDAO.getTaskInfo(taskId);

        List<TaskTagRule> tagRuleList = TaskTagRuleDAO.getTagRuleList(taskId);


        System.out.println(tagInfo.getTagName());
        System.out.println(taskInfo.getTaskName());
        System.out.println(tagRuleList);


        //2、  建表问题
        //          最好不要手动建表 create table
        //思考，所有标签建一个表 ，每个标签一个表？ 都可以
        //   每个标签一个表
        //   create table  $tableName   tag_code
        //   (uid string ,tag_value $fieldType)  tag_info.tag_value_type  1 => bigint 2 => decimal 3 ==>string 4 ==>string
        //   分区 partitioned by  (dt string)
        //  压缩？ 格式？   文本行格式
        //  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
        //  location   hdfs://xxx/xx/xx/库名/表名  配置文件地址/库名/表名

        String tableName = tagInfo.getTagCode().toLowerCase();
        String fieldType="";
        if (tagInfo.getTagValueType().equals(TAG_VALUE_TYPE_LONG)){
            fieldType="bigint";
        }else    if (tagInfo.getTagValueType().equals(TAG_VALUE_TYPE_DECIMAL)){
            fieldType="decimal";
        }else    if (tagInfo.getTagValueType().equals(TAG_VALUE_TYPE_STRING)){
            fieldType="string";
        } else   if (tagInfo.getTagValueType().equals(TAG_VALUE_TYPE_DATE)){
            fieldType="string";
        }

        Properties properties = MyPropertiesUtil.load("config.properties");
        String hdfsPath = properties.getProperty("hdfs-store.path");
        String dwDbName = properties.getProperty("data-warehouse.dbname");
        String upDbName = properties.getProperty("user-profile.dbname");

        String createTableSQL=" create table  if not exists "+upDbName+"."+tableName+
                "           (uid string ,tag_value "+fieldType+")   " +
                "            partitioned by  (dt string) " +
                "            comment   '"+tagInfo.getTagName()+"'"+
                "           ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'\n" +
                "          location   '"+hdfsPath+"/"+upDbName+"/"+tableName+"'";
        System.out.println(createTableSQL);


       sparkSession.sql(createTableSQL);


        //3、  查询电商离线数仓
        //                拼接组合 sql     --》 insert overwrite select ....
        //  select uid,  case  query_value  when 'F' then '女' when 'M' then '男' ...  end  tag_value from (
       //  select  id as uid,  nvl(gender,'U')  as  query_value
        // from dim_user_zip where dt='9999-12-31' ) tv

        //作业： 1 如果没有四级标签，则直接用query_value作为tag_value
        //      2 把$dt替换为busi_date
        String tagValueSQL=null;
        if(tagRuleList.size()>0){
            List<String> whenThenList = tagRuleList.stream().map(tagRule ->
                    "when '" + tagRule.getQueryValue() + "' then '" + tagRule.getSubTagValue()+"'"
            ).collect(Collectors.toList());
            String whenThenSQL = StringUtils.join(whenThenList," ");

            String caseSQL="case query_value "+whenThenSQL+" end ";
            tagValueSQL=caseSQL;
        }else{
            tagValueSQL="query_value";
        }


        String selectSQL="select uid ,"+tagValueSQL+ " as  tag_value from ("+taskInfo.getTaskSql()+") tv";
        //  作业：    2 把$dt替换为busi_date
        String selectSQLWithBusiDate = selectSQL.replace("$dt", busiDate);

        System.out.println(selectSQLWithBusiDate);

        //4、  生成标签信息，写入画像库

        String insertSQL=" insert overwrite table "+upDbName+"."+tableName+" partition( dt='"+busiDate+"' )"+selectSQLWithBusiDate;

        System.out.println(insertSQL);

        sparkSession.sql("use "+dwDbName);
        sparkSession.sql(insertSQL);



    }
}
